import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.LinkedList;
import java.util.List;
import java.util.TimeZone;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.services.autoscaling.AmazonAutoScalingClient;
import com.amazonaws.services.cloudwatch.AmazonCloudWatchClient;
import com.amazonaws.services.cloudwatch.model.Datapoint;
import com.amazonaws.services.cloudwatch.model.Dimension;
import com.amazonaws.services.cloudwatch.model.GetMetricStatisticsRequest;
import com.amazonaws.services.cloudwatch.model.GetMetricStatisticsResult;
import com.amazonaws.services.ec2.AmazonEC2;
import com.amazonaws.services.ec2.AmazonEC2Client;
import com.amazonaws.services.ec2.model.AssociateAddressRequest;
import com.amazonaws.services.ec2.model.CreateImageRequest;
import com.amazonaws.services.ec2.model.CreateImageResult;
import com.amazonaws.services.ec2.model.CreateTagsRequest;
import com.amazonaws.services.ec2.model.DescribeInstancesRequest;
import com.amazonaws.services.ec2.model.DescribeInstancesResult;
import com.amazonaws.services.ec2.model.Placement;
import com.amazonaws.services.ec2.model.RunInstancesRequest;
import com.amazonaws.services.ec2.model.RunInstancesResult;
import com.amazonaws.services.ec2.model.Tag;
import com.amazonaws.services.ec2.model.TerminateInstancesRequest;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.S3Object;


public class Head {
	
	/**
	 * @param args
	 */
	static AmazonEC2      ec2;
    static AmazonS3       s3;
    static AmazonCloudWatchClient cloudWatch;
    static AmazonAutoScalingClient autoScale;
    
    static String bucketName = "wangzitong";
    static String key = "client";
    static private boolean c[]= {false,false,false,false,false,false,false,false,false,false};
    static private double cpu[] = {-1,-1,-1,-1,-1,-1,-1,-1,-1,-1};
    static private String insIds[] = new String[10];
    static private int busy[] = {0,0,0,0,0,0,0,0,0,0};	//flag of whether a client is busy
    static private int idle[] = {0,0,0,0,0,0,0,0,0,0};	//count of consecutive idle times of a client
    static private boolean lock[] = {false,false,false,false,false,false,false,false,false,false};	//lock the client to avoid terminating when create images.
    static private String ip[] = new String[10];
    static private int working = 0;	//# of working clients 
    static int fileport = 20000;
    static String imageId_client;
    
    private static void init() throws Exception {
        AWSCredentials credentials = new PropertiesCredentials(
                Head.class.getResourceAsStream("AwsCredentials.properties"));

        ec2 = new AmazonEC2Client(credentials);
        s3  = new AmazonS3Client(credentials);
        cloudWatch = new AmazonCloudWatchClient(credentials);
        autoScale = new AmazonAutoScalingClient(credentials);
    }
    
    public static boolean get_client(int i) {
    	return c[i];
    }
    
    public static double get_cpu(int i) {
    	if(c[i]) return cpu[i];
    	return -1;
    }
    
    public static int get_busy(int i) {
    	return busy[i];
    }
    
    public static int get_idle(int i) {
    	return idle[i];
    }
    
    public static int get_working() {
    	return working;
    }
    
    private static class CreateInstance implements Runnable {

		@Override
		public void run() {
			// TODO Auto-generated method stub
			try {
				System.out.println("Create instance");
				RunInstancesRequest rirnew = new RunInstancesRequest(imageId_client, 1, 1);
	            Placement placenew = new Placement("us-east-1c");
	            rirnew.withKeyName("MyKey");
	            rirnew.setMonitoring(true);
	            rirnew.withSecurityGroups("MyGroup");
	            rirnew.withInstanceType("t1.micro");
	            rirnew.withPlacement(placenew);
	            RunInstancesResult resultnew = ec2.runInstances(rirnew);
	            String ins_new = resultnew.getReservation().getInstances().get(0).getInstanceId();
	            
	            Thread.sleep(30000);
	            DescribeInstancesRequest dir = new DescribeInstancesRequest();
	            dir.withInstanceIds(ins_new);
	            DescribeInstancesResult diresult = ec2.describeInstances(dir);
	            String ip_new = diresult.getReservations().get(0).getInstances().get(0).getPublicIpAddress();
	            for(int j=0;j<10;j++) {
	            	if(!c[j]) {
	            		c[j] = true;
	            		insIds[j] = ins_new;
	            		busy[j] = 0;
	            		idle[j] = 0;
	            		ip[j] = ip_new;
	            		working++;
	            		break;
	            	}
	            }
			} catch(Exception e) {
				e.printStackTrace();
			}			
		}    	
    }
    
    private static class CreateImage implements Runnable {
    	int i;
    	
    	CreateImage(int i) {
    		this.i = i;
    	}
    	
		@Override
		public void run() {
			// TODO Auto-generated method stub
			try {
				System.out.println("Create Image for client "+i);
				Thread.sleep(5000);	//five seconds for client to deploy new service
				CreateImageResult cir = ec2.createImage(new CreateImageRequest(insIds[i],insIds[i]));
				imageId_client = cir.getImageId();
				s3.deleteObject(bucketName, key);
				File file = File.createTempFile(key, ".txt");
	            file.deleteOnExit();
	            Writer wr = new OutputStreamWriter(new FileOutputStream(file));
	            wr.write(imageId_client+"\r\n");
	            wr.close();            
	            s3.putObject(bucketName, key, file);
	            Thread.sleep(40000);
	            lock[i] = false;
			} catch(Exception e) {
				e.printStackTrace();
			}
		}    	
    }
    
    private static class ReceiveMsg implements Runnable {	//not tested
    	ServerSocket mysoc = null;
    	
    	ReceiveMsg() {
    		try {
    			this.mysoc = new ServerSocket(fileport);
    		} catch(Exception e) {
    			e.printStackTrace();
    		}
    	}
    	
		@Override
		public void run() {
			// TODO Auto-generated method stub
			try {
				while(true) {
					Socket clientsoc = mysoc.accept();
					BufferedReader inFromClient = new BufferedReader(new InputStreamReader(clientsoc.getInputStream()));
					while(inFromClient.ready()) {
						String msg = inFromClient.readLine();
						System.out.println("Received from "+clientsoc.getInetAddress()+": "+msg);
					}
					inFromClient.close();
					clientsoc.close();
				}
			} catch(Exception e) {
				e.printStackTrace();
			}			
		}    	
    }
    
    private static double watch(String insId) {
    	System.out.println("Watching "+insId);
    	GetMetricStatisticsRequest statRequest = new GetMetricStatisticsRequest();
        statRequest.setNamespace("AWS/EC2"); //namespace
		statRequest.setPeriod(60); //period of data
		ArrayList<String> stats = new ArrayList<String>();
		stats.add("Average"); 
		statRequest.setStatistics(stats);
		statRequest.setMetricName("CPUUtilization");
		GregorianCalendar calendar = new GregorianCalendar(TimeZone.getTimeZone("UTC"));
		calendar.add(GregorianCalendar.SECOND, -1); // 1 second ago
		Date endTime = calendar.getTime();
		calendar.add(GregorianCalendar.SECOND, -61); // 1 minutes ago
		Date startTime = calendar.getTime();
		statRequest.setStartTime(startTime);
		statRequest.setEndTime(endTime);
		ArrayList<Dimension> dimensions = new ArrayList<Dimension>();
		dimensions.add(new Dimension().withName("InstanceId").withValue(insId));
		statRequest.setDimensions(dimensions);
    
		GetMetricStatisticsResult statResult = cloudWatch.getMetricStatistics(statRequest);
		
		List<Datapoint> dataList = statResult.getDatapoints();
		Double CPU = null;
		for(Datapoint data:dataList) {
			CPU = data.getAverage();
			System.out.println(insId+": "+CPU);
			return CPU.doubleValue();
		}
    	return -1;
    }
    
    public static void file() throws Exception {
    	String dir = "f:\\COMS E6998\\project\\uploaded\\";
    	File f = new File(dir+"record.txt");
    	if(f.exists()) {
    		FileReader fr = new FileReader(f);
    		BufferedReader br = new BufferedReader(fr);
    		while(br.ready()) {
    			String filename = br.readLine();
    			String command = br.readLine();
    			if(filename.equals("") && !command.equals("")) {	//only command, send to not busy client
    				for(int i=0;i<10;i++) {
    					if(c[i] && busy[i]==0) {
    						Socket soc = new Socket(ip[i],fileport);
    						FileTransferer.send(soc, null, command);
    						break;
    					}
    				}
    			} else if(!filename.equals("")) {
    				System.out.println("Send file: "+filename);
    				File file = new File(dir+filename);
    				for(int i=0;i<10;i++) {
    					if(c[i]) {
    						Socket soc = new Socket(ip[i],fileport);
    						FileTransferer.send(soc, file, command);
    					}
    				}
    				for(int i=0;i<10;i++) {
    					if(c[i]) {
    						lock[i] = true;
    						new Thread(new CreateImage(i)).start();
    						break;
    					}
    				}
    			}
    		}
    		br.close();
    		f.delete();
    	}
    }
    
    
    
	public static void main(String args[]) throws Exception {
		// TODO Auto-generated method stub
		init();
		try {
			System.out.println("Read configuration file.");
//			S3Object object = s3.getObject(new GetObjectRequest(bucketName, key1));
//            BufferedReader reader = new BufferedReader(new InputStreamReader(object.getObjectContent()));
//            String imageId_head = reader.readLine();
//            String ip = reader.readLine();
//            System.out.println(imageId_head+" "+ip);
            S3Object object = s3.getObject(new GetObjectRequest(bucketName, key));
            BufferedReader reader = new BufferedReader(new InputStreamReader(object.getObjectContent()));
            imageId_client = reader.readLine();
            System.out.println(imageId_client);
            
            //System.out.println("Create instances");  
            new Thread(new CreateInstance()).start();            
            
//            CreateTagsRequest ctr = new CreateTagsRequest();
//            ctr.withResources(ins_head).withTags(new Tag("Name","Head node"));
//            ec2.createTags(ctr);
            
            System.out.println("Wait for 40 seconds...");
            Thread.sleep(40000);
            
            System.out.println("Start Receiving at port "+fileport);
            new Thread(new ReceiveMsg()).start();
            
            System.out.println("Start monitoring");
            while(true) {
            	for(int i=0;i<10;i++) {
            		if(c[i]) {
            			cpu[i] = watch(insIds[i]);
            			if(cpu[i] > 45) {
            				System.out.println("client "+i+" CPU: "+cpu[i]);
            				busy[i] = 1;
            				idle[i] = 0;
            				int busycpu = 0;
            				for(int j=0;j<10;j++)busycpu += busy[j];
            				if(busycpu==working && working < 11) {
            					//System.out.println("Create new client"); 
            					new Thread(new CreateInstance()).start();
            				}
            			} else if(cpu[i] > 15) {
            				System.out.println("client "+i+" CPU: "+cpu[i]);
            				idle[i] = 0;
            			} else if(cpu[i] >= 0) {
            				System.out.println("client "+i+" CPU: "+cpu[i]);
            				busy[i] = 0;
            				idle[i]++;
            				if(idle[i]>4 && !lock[i]) {	//idle for a long time, terminate if have other not busy clients
            					int busycpu = 0;
                				for(int j=0;j<10;j++)busycpu += busy[j];
                				if(working-busycpu > 1) {
                					System.out.println("Terminate idle client "+i);
                					List<String> instanceIds = new LinkedList<String>();
                					instanceIds.add(insIds[i]);
                					ec2.terminateInstances(new TerminateInstancesRequest(instanceIds));
                					c[i] = false;
                					idle[i] = 0;
                					cpu[i] = -1;
                					working--;
                				}
            				}
            			}
            		}
            	}
            	file();
            	Thread.sleep(30000);
            }
               
		} catch(Exception e) {
			e.printStackTrace();
		}
	}

}
